package com.yeziji.dataCenter.listener.rabbitMq;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson2.JSONObject;
import com.yeziji.common.business.dataMessage.base.DataMessage;
import com.yeziji.common.business.dataMessage.service.DataMessageRecordService;
import com.yeziji.common.rabbitMq.RabbitMqTopicEnum;
import com.yeziji.common.rabbitMq.RabbitMqTopicHandler;
import com.yeziji.constant.DataMessageTopicEnum;
import com.yeziji.dataCenter.business.dataSource.service.DataOperateService;
import com.yeziji.utils.RabbitMqUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * 平台新增数据消息监听器
 *
 * @author hwy
 * @since 2024/10/25 14:36
 **/
@Slf4j
@Component
public class DataMessageInsertListener extends RabbitMqTopicHandler {
    @Resource
    private DataOperateService dataOperateService;
    @Resource
    private DataMessageRecordService dataMessageRecordService;

    @Override
    public RabbitMqTopicEnum getTopicEnum() {
        return DataMessageTopicEnum.PLATFORM_DATA_INSERT_SYNC;
    }

    @Override
    public void process(Message message, String body) {
        DataMessage dataMessage = JSONObject.parseObject(body, DataMessage.class);
        dataOperateService.execute(dataMessage);
    }

    @Override
    public void deadProcess(Message message, String body) {
        log.info("处理新增死信消息: {} -- {}", message, body);
        DataMessage dataMessage = JSONObject.parseObject(body, DataMessage.class);
        String clientId = dataMessage.getClientId();

        // 将死信转发给对应的平台死信队列
        MessageProperties messageProperties = message.getMessageProperties();
        if (messageProperties != null && clientId != null) {
            String receivedExchange = messageProperties.getReceivedExchange();
            String platformRouteKey = RabbitMqUtils.getAsRabbitPlatformKey(clientId, messageProperties.getReceivedRoutingKey());
            dataMessageRecordService.sendTo(receivedExchange, platformRouteKey, dataMessage, null);
        }
    }

    @Override
    public void confirmAck(CorrelationData correlationData, String message) {
        log.info("发送消息成功: {}", message);
    }

    @Override
    public void confirmUnAck(CorrelationData correlationData, String message) {
        log.info("发送消息失败: {}", message);
    }

    @Override
    public void returnedProcess(ReturnedMessage returnedMessage) {
        log.info("消息异常: {}", returnedMessage);
    }
}